package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	//	"bytes"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"
	"bytes"

	"fmt"
	"6.5840/labgob"
	"6.5840/labrpc"
)

const(
	FOLLOWER = 0
	CANDIDATE = 1
	LEADER = 2
)

type Log struct {
	Term    int
	Index   int
	Command interface{}
}

// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in part 2D you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh, but set CommandValid to false for these
// other uses.
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int

	// For 2D:
	SnapshotValid bool
	Snapshot      []byte
	SnapshotTerm  int
	SnapshotIndex int
}

// A Go object implementing a single Raft peer.
type Raft struct {
	mu            sync.Mutex          // Lock to protect shared access to this peer's state
	peers         []*labrpc.ClientEnd // RPC end points of all peers
	persister     *Persister          // Object to hold this peer's persisted state
	me            int                 // this peer's index into peers[]
	dead          int32               // set by Kill()

	// Your data here (2A, 2B, 2C).
	// Look at the paper's Figure 2 for a description of what
	// state a Raft server must maintain.
	
	// Persistant state
	CurrentTerm     int
    VotedFor        int
    Log           []Log

	LastIncludeIndex int
	LastIncludeTerm  int

	SnapshotMirror      []byte

	state           int

	// Volatile state for all servers
	commit_index    int
	last_applied    int
	
	// Volatile state for leaders
	next_index      []int
	match_index     []int

	leader_detected bool
	apply_channel   chan ApplyMsg
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	var term int
	var isleader bool
	// Your code here (2A).

	term = rf.CurrentTerm
	isleader = rf.state == LEADER

	return term, isleader
}

// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
// before you've implemented snapshots, you should pass nil as the
// second argument to persister.Save().
// after you've implemented snapshots, pass the current snapshot
// (or nil if there's not yet a snapshot).
func (rf *Raft) persist() {
	// Your code here (2C).
	// Example:
	// w := new(bytes.Buffer)
	// e := labgob.NewEncoder(w)
	// e.Encode(rf.xxx)
	// e.Encode(rf.yyy)
	// raftstate := w.Bytes()
	// rf.persister.Save(raftstate, nil)
	w := new(bytes.Buffer)
	e := labgob.NewEncoder(w)
	e.Encode(rf.CurrentTerm)
	e.Encode(rf.VotedFor)
	e.Encode(rf.Log)
	raftstate := w.Bytes()
	rf.persister.Save(raftstate, rf.SnapshotMirror)
}


// restore previously persisted state.
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	// Your code here (2C).
	// Example:
	// r := bytes.NewBuffer(data)
	// d := labgob.NewDecoder(r)
	// var xxx
	// var yyy
	// if d.Decode(&xxx) != nil ||
	//    d.Decode(&yyy) != nil {
	//   error...
	// } else {
	//   rf.xxx = xxx
	//   rf.yyy = yyy
	// }
	rf.mu.Lock()
	defer rf.mu.Unlock()
	r := bytes.NewBuffer(data)
	d := labgob.NewDecoder(r)
	if d.Decode(&rf.CurrentTerm) != nil || d.Decode(&rf.VotedFor) != nil || d.Decode(&rf.Log) != nil {
		panic("server failed to restored")
	}
	// It can be null, so do not need to check it
	rf.SnapshotMirror = rf.persister.ReadSnapshot()
	if (rf.LastIncludeIndex < rf.Log[0].Index) {
		rf.LastIncludeIndex = rf.Log[0].Index
		rf.LastIncludeTerm  = rf.Log[0].Term
	}
}


// the service says it has created a snapshot that has
// all info up to and including index. this means the
// service no longer needs the log through (and including)
// that index. Raft should now trim its log as much as possible.
func (rf *Raft) Snapshot(index int, snapshot []byte) {
	// Your code here (2D).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if (index < rf.Log[0].Index) {
		// The snapshot is out of date,we simply ignore it
		return
	} else if (index > rf.Log[len(rf.Log)-1].Index) {
		// We can not set lastIncludeTerm,also ignore it
		fmt.Printf("snapshot index is greater than any existing logs\n")
		return
	}
	direct_index := index - rf.Log[0].Index
	rf.LastIncludeIndex = rf.Log[direct_index].Index
	rf.LastIncludeTerm  =  rf.Log[direct_index].Term
	rf.SnapshotMirror         = snapshot
	rf.Log = rf.Log[direct_index:]

	if (rf.commit_index < index) {
		rf.commit_index = index
	}
	if (rf.match_index[rf.me] < index) {
		rf.match_index[rf.me] = index
	}
}


// example RequestVote RPC arguments structure.
// field names must start with capital letters!
type RequestVoteArgs struct {
	// Your data here (2A, 2B).
	Term         int
	CandidateId  int
	LastLogIndex int
	LastLogTerm  int
}

// example RequestVote RPC reply structure.
// field names must start with capital letters!
type RequestVoteReply struct {
	// Your data here (2A).
	Term      int
	IsGranted bool
}

type AppendEntriesArgs struct {
	Term              int
	LeaderId          int
	PrevLogIndex      int
	PrevLogTerm       int
	Entries           []Log
	LeaderCommit      int
}

type AppendEntriesReply struct {
	Term       int
	InferIndex int
	TraceTerm  int
	Success    bool
}

type InstallSnapshotRPCArgs struct {
	Term              int
	LeaderId          int
	LastIncludeIndex  int
	LastIncludeTerm   int

	// For lab2D not used
	Offset            int

	Data              []byte

	// For lab2D, not used
	Done              bool
}

type InstallSnapshotRPCReply struct {
	Term              int
}

func (rf *Raft) check_term(other_term int) {
	if (other_term > rf.CurrentTerm) {
		rf.state = FOLLOWER
		rf.VotedFor = -1
		rf.CurrentTerm = other_term
		rf.persist()
	}
}

// example RequestVote RPC handler.
func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	reply.Term = rf.CurrentTerm
	if (rf.CurrentTerm > args.Term) {
		reply.IsGranted = false
	} else {
		granded := false
		rf.check_term(args.Term)
		if (rf.VotedFor == -1) {
			last_term := rf.Log[len(rf.Log)-1].Term
			if (args.LastLogTerm == last_term) {
				if (args.LastLogIndex >= rf.Log[len(rf.Log)-1].Index) {
					granded = true
					rf.VotedFor = args.CandidateId
				}
			} else if (args.LastLogTerm > last_term) {
				granded = true
				rf.VotedFor = args.CandidateId
			}
		}
		reply.IsGranted = granded
	}
}

func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) {
	// Your code here (2A, 2B).
	rf.mu.Lock()
	defer rf.mu.Unlock()
	reply.Term = rf.CurrentTerm
	reply.Success = false
	reply.TraceTerm = -1
	if (rf.CurrentTerm > args.Term) {
		// Do not admit it as leader, just tell it your term
		reply.InferIndex = 1
		return 
	}
	if (rf.state != FOLLOWER || rf.VotedFor != -1 || rf.CurrentTerm < args.Term) {
		rf.state = FOLLOWER
		rf.VotedFor = -1
		rf.CurrentTerm = args.Term
		rf.persist()
	}
	rf.leader_detected = true
	real_log_prev_index := rf.calculate_log_arr_index(args.PrevLogIndex)
	last_index := len(rf.Log) - 1
	if (real_log_prev_index > last_index) {
		// This case we miss some logs, tell leader to change it's correspond next_index
		reply.InferIndex = rf.get_last_log_index()+1
		return
    }
	if (real_log_prev_index < 0) {
		reply.InferIndex = rf.LastIncludeIndex
		return
	}
	if (rf.Log[real_log_prev_index].Term < args.PrevLogTerm) {
		// In this case, we can not infer how to find index match leader term, simply reduce by one
		// TODO: Update request, reply structure to let leader side optimize this situation
		reply.InferIndex = args.PrevLogIndex-1
		reply.TraceTerm = rf.Log[real_log_prev_index].Term
		return
	} else if (rf.Log[real_log_prev_index].Term > args.PrevLogTerm) {
		// Find the first log index where log's term is equal or smaller than leader side's index
		for ;rf.Log[real_log_prev_index].Term > args.PrevLogTerm && real_log_prev_index>0;real_log_prev_index--{
		}
		reply.InferIndex = rf.LastIncludeIndex+real_log_prev_index+1
		return
	}
	reply.Success = true
	real_log_prev_index++
	rf.Log = append(rf.Log[:real_log_prev_index], args.Entries...)
	rf.persist()
	last_index = rf.Log[len(rf.Log)-1].Index
	if (args.LeaderCommit > rf.commit_index) {
		if (args.LeaderCommit > last_index) {
			rf.commit_index = last_index
		} else {
			rf.commit_index = args.LeaderCommit
		}
	}
}


func (rf *Raft) InstallSnapshotRPC(args *InstallSnapshotRPCArgs, reply *InstallSnapshotRPCReply) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	reply.Term = rf.CurrentTerm
	if (args.Term < rf.CurrentTerm) {
		return
	}
	rf.SnapshotMirror = args.Data
	rf.LastIncludeIndex = args.LastIncludeIndex
	rf.LastIncludeTerm  = args.LastIncludeTerm
	rf.Log = make([]Log, 1)
	rf.Log[0].Term = rf.LastIncludeTerm
	rf.Log[0].Index = rf.LastIncludeIndex

	if (rf.commit_index < rf.LastIncludeIndex) {
		rf.commit_index = rf.LastIncludeIndex
	}
}

// example code to send a RequestVote RPC to a server.
// server is the index of the target server in rf.peers[].
// expects RPC arguments in args.
// fills in *reply with RPC reply, so caller should
// pass &reply.
// the types of the args and reply passed to Call() must be
// the same as the types of the arguments declared in the
// handler function (including whether they are pointers).
//
// The labrpc package simulates a lossy network, in which servers
// may be unreachable, and in which requests and replies may be lost.
// Call() sends a request and waits for a reply. If a reply arrives
// within a timeout interval, Call() returns true; otherwise
// Call() returns false. Thus Call() may not return for a while.
// A false return can be caused by a dead server, a live server that
// can't be reached, a lost request, or a lost reply.
//
// Call() is guaranteed to return (perhaps after a delay) *except* if the
// handler function on the server side does not return.  Thus there
// is no need to implement your own timeouts around Call().
//
// look at the comments in ../labrpc/labrpc.go for more details.
//
// if you're having trouble getting RPC to work, check that you've
// capitalized all field names in structs passed over RPC, and
// that the caller passes the address of the reply struct with &, not
// the struct itself.
func (rf *Raft) sendRequestVote(server int, args *RequestVoteArgs, reply *RequestVoteReply) bool {
	ok := rf.peers[server].Call("Raft.RequestVote", args, reply)
	return ok
}

func (rf *Raft) sendAppendEntriesRPC(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool {
	ok := rf.peers[server].Call("Raft.AppendEntries", args, reply)
	return ok
}

func (rf *Raft) sendInstallSnapshotRPC(server int, args *InstallSnapshotRPCArgs, reply *InstallSnapshotRPCReply) bool {
	ok := rf.peers[server].Call("Raft.InstallSnapshotRPC", args, reply)
	return ok
}


// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()

	index := -1
	term := -1
	isLeader := true

	// Your code here (2B).
	if (rf.state == LEADER) {
		log := Log{}
		log.Command = command
		log.Term    = rf.CurrentTerm
		log.Index   = rf.Log[len(rf.Log)-1].Index+1
		rf.Log = append(rf.Log, log)
		rf.next_index[rf.me] = rf.Log[len(rf.Log)-1].Index+1
		rf.match_index[rf.me] = rf.Log[len(rf.Log)-1].Index
		index = rf.match_index[rf.me]
		term  = log.Term
		rf.persist()
	} else {
		isLeader = false
	}


	return index, term, isLeader
}

// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) calculate_log_arr_index(index int) int {
	return index - rf.LastIncludeIndex
}

func (rf *Raft) get_last_log_index() int {
	return rf.Log[len(rf.Log)-1].Index
}

func (rf *Raft) ticker() {
	for rf.killed() == false {
		// Your code here (2A)
		// Check if a leader election should be started.
		rf.mu.Lock()
		if (rf.state == CANDIDATE || (rf.state == FOLLOWER && !rf.leader_detected && rf.VotedFor == -1)) {
			time_wait_channel := make(chan bool, 1)
			go rf.elect(time_wait_channel)
			ms := 100 + (rand.Int63() % 200)
		    time.Sleep(time.Duration(ms) * time.Millisecond)
			time_wait_channel <- true
		} else {
			rf.leader_detected = false
			rf.mu.Unlock()
			// pause for a random amount of time between 250 and 550
		    // milliseconds.
			ms := 100 + (rand.Int63() % 200)
		    time.Sleep(time.Duration(ms) * time.Millisecond)
		}
	}
}

// Launch a leader election
func (rf *Raft) elect(time_wait chan bool) {
	rf.CurrentTerm++
	rf.VotedFor = rf.me
	rf.state = CANDIDATE
	rf.persist()
	rf.mu.Unlock()
	results_chan := make(chan bool, len(rf.peers))
	for i := 0; i < len(rf.peers); i++ {
		if (rf.me != i) {
			go func(res_chan chan bool, rf *Raft, i int) {
				rf.mu.Lock()
				req := RequestVoteArgs{}
	            req.Term = rf.CurrentTerm
	            req.CandidateId = rf.me
	            req.LastLogIndex = rf.Log[len(rf.Log)-1].Index
	            req.LastLogTerm = rf.Log[len(rf.Log)-1].Term
				res := RequestVoteReply{}
				rf.mu.Unlock()
				rpc_ok := rf.sendRequestVote(i, &req, &res)
				if (!rpc_ok) {
					return
				}
				rf.mu.Lock()
				rf.check_term(res.Term)
				if (rpc_ok && res.IsGranted) {
					results_chan <- true
				} else {
					results_chan <- false
				}
				rf.mu.Unlock()
			}(results_chan, rf, i);
		}
	}
	election_timouted := false
	votes_num := 1
	for ; votes_num <= len(rf.peers)/2 && !rf.leader_detected && !election_timouted; {
		for {
			nothing_to_read := false
			select {
			case aggreed := <-results_chan:
				if (aggreed) {
					votes_num++;
				}
			default:
				nothing_to_read = true
			}
			if (nothing_to_read) {
				break
			}
		}
		if (votes_num > len(rf.peers)/2) {
			break
		}
		select {
		case tt := <- time_wait:
			if (tt) {
				election_timouted = true
			}
		default:
			time.Sleep(time.Duration(10) * time.Millisecond)
		}
	}
	if (votes_num > len(rf.peers)/2) {
		rf.mu.Lock()
		rf.state = LEADER
		for init_index := 0; init_index < len(rf.peers); init_index++ {
			rf.next_index[init_index]  = rf.Log[len(rf.Log)-1].Index + 1
			rf.match_index[init_index] = 0
		}
		rf.mu.Unlock()
		go rf.leader_heart_beat()
	}
}

func (rf *Raft) leader_heart_beat() {
	for rf.state == LEADER {
		for i := 0; i < len(rf.peers); i++{
			if (i != rf.me) {
				go func(rf *Raft, index int) {
					args := AppendEntriesArgs{}
					reply := AppendEntriesReply{}
					rf.mu.Lock();
					if (rf.state != LEADER) {
						rf.mu.Unlock()
						return
					}
					if (rf.next_index[index] <= rf.LastIncludeIndex) {
						rf.make_install_snapshot_rpc_call_for_append_entries(index)
						rf.mu.Unlock()
						return
					}
					if (rf.next_index[index] > rf.get_last_log_index()+1) {
						rf.next_index[index] = rf.get_last_log_index()+1
					}
					args.Term = rf.CurrentTerm
					args.LeaderId = rf.me
					args.PrevLogIndex = rf.next_index[index] - 1
					args.PrevLogTerm  = rf.Log[args.PrevLogIndex-rf.LastIncludeIndex].Term
					args.LeaderCommit = rf.commit_index
					append_index := args.PrevLogIndex+1-rf.LastIncludeIndex
					total_log_length := len(rf.Log)
					args.Entries = make([]Log, total_log_length - append_index)
					preserved_next_index := total_log_length+rf.LastIncludeIndex
					for cursor := append_index; cursor < total_log_length; cursor++ {
						args.Entries[cursor - append_index] = rf.Log[cursor]
					}
					rf.mu.Unlock()
					res := rf.sendAppendEntriesRPC(index, &args, &reply)
					if (!res) {
						return
					}
					rf.mu.Lock()
					defer rf.mu.Unlock()
					if (reply.Success) {
						if (preserved_next_index > rf.get_last_log_index()+1) {
							fmt.Printf("expected error in success append entries reply..\n")
						}
						rf.next_index[index]  = preserved_next_index
						rf.match_index[index] = preserved_next_index-1
					} else {
						if (rf.CurrentTerm < reply.Term) {
							rf.check_term(reply.Term)
						} else {
							if (reply.TraceTerm != -1) {
								for rf.next_index[index] > rf.LastIncludeIndex && rf.Log[rf.next_index[index]-1-rf.LastIncludeIndex].Term > reply.TraceTerm {
									rf.next_index[index]--
								}
								return
							}
							if (preserved_next_index < reply.InferIndex) {
								fmt.Printf("don not expected to happens, but appear yet... in infer index")
							}
						    rf.next_index[index] = reply.InferIndex
						}
					}
				}(rf, i)
			}
		}
		time.Sleep(time.Duration(50) * time.Millisecond)
	}
}

func (rf *Raft) make_install_snapshot_rpc_call(server int) {
	rf.mu.Lock()
	args := InstallSnapshotRPCArgs{}
	reply := InstallSnapshotRPCReply{}
	args.Term = rf.CurrentTerm
	args.LeaderId = rf.me
	args.LastIncludeIndex = rf.LastIncludeIndex
	args.LastIncludeTerm  = rf.LastIncludeTerm
	args.Data             = rf.SnapshotMirror
	rf.mu.Unlock()
	rf.sendInstallSnapshotRPC(server, &args, &reply)
	rf.mu.Lock()
	if (reply.Term > rf.CurrentTerm) {
		rf.check_term(reply.Term)
	}
	if (rf.LastIncludeIndex+1 > rf.get_last_log_index()) {
		fmt.Printf("??????\n")
	}
	rf.next_index[server] = rf.LastIncludeIndex+1
	rf.match_index[server] = rf.LastIncludeIndex
	rf.mu.Unlock()
}

func (rf *Raft) make_install_snapshot_rpc_call_for_append_entries(server int) {
	args := InstallSnapshotRPCArgs{}
	reply := InstallSnapshotRPCReply{}
	args.Term = rf.CurrentTerm
	args.LeaderId = rf.me
	args.LastIncludeIndex = rf.LastIncludeIndex
	args.LastIncludeTerm  = rf.LastIncludeTerm
	args.Data             = rf.SnapshotMirror
	rf.mu.Unlock()
	rf.sendInstallSnapshotRPC(server, &args, &reply)
	rf.mu.Lock()
	if (reply.Term > rf.CurrentTerm) {
		rf.check_term(reply.Term)
		return
	}
	if (rf.LastIncludeIndex > rf.get_last_log_index()) {
		fmt.Printf("??????\n")
	}
	rf.next_index[server] = rf.LastIncludeIndex+1
	rf.match_index[server] = rf.LastIncludeIndex
}

func (rf *Raft) leader_check_snapshot_install() {
	for i:=0; i < len(rf.peers); i++ {
		if (i != rf.me && rf.next_index[i] <= rf.LastIncludeIndex) {
			go rf.make_install_snapshot_rpc_call(i)
		}
	}
}

func (rf *Raft) leader_check_next_index_update() {
	for idx := rf.get_last_log_index(); idx > rf.commit_index; idx--{
		sum := 0
		for i := 0; i < len(rf.peers); i++{
			if (rf.match_index[i] >= idx) {
				sum++
			}
		}
		if (sum > len(rf.peers)/2) {
			rf.commit_index = idx
			return
		}
	}
}

func (rf *Raft) try_apply_log() {
	if (rf.last_applied < rf.LastIncludeIndex) {
		msg := ApplyMsg{}
		msg.CommandValid = false
		msg.SnapshotValid = true
		msg.SnapshotIndex = rf.LastIncludeIndex
		msg.SnapshotTerm = rf.LastIncludeTerm
		msg.Snapshot        = rf.SnapshotMirror

		rf.apply_channel <- msg

		rf.last_applied = rf.LastIncludeIndex
	}
	if (rf.last_applied < rf.commit_index && rf.Log[rf.commit_index - rf.LastIncludeIndex].Term == rf.CurrentTerm) {
		idx := rf.last_applied+1
		msg := ApplyMsg{}
		msg.CommandValid = true
		msg.CommandIndex = idx
		msg.Command      = rf.Log[idx-rf.LastIncludeIndex].Command
		rf.apply_channel <- msg
		rf.last_applied++
	}
	/*
	fmt.Printf("server %v with last_applied:%v and last_commit:%v logs:%v\n", rf.me, rf.last_applied, rf.commit_index, rf.Log)
	if (rf.state == LEADER) {
		fmt.Printf("leader next_index:%v match_index:%v\n", rf.next_index, rf.match_index)
	}
	*/
}

func (rf *Raft) check_commit_and_try_apply() {
	for rf.killed() == false {
		rf.mu.Lock()
		if (rf.state == LEADER) {
			rf.leader_check_next_index_update()
		}
		if (rf.commit_index >= rf.get_last_log_index()+1) {
			rf.commit_index = rf.get_last_log_index()
			rf.mu.Unlock()
			continue
		}
		rf.try_apply_log()
		rf.mu.Unlock()
		time.Sleep(time.Duration(10) * time.Millisecond)
	}
}

// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.CurrentTerm  = 0
	rf.VotedFor     = -1
	rf.Log          = make([]Log, 1)
	rf.Log[0].Term  = 0
	rf.Log[0].Index = 0

	rf.state        = FOLLOWER

	rf.commit_index = 0
	rf.last_applied = 0

	rf.LastIncludeIndex = 0
	rf.LastIncludeTerm  = 0
	
	rf.next_index   = make([]int, len(peers))
	rf.match_index  = make([]int, len(peers))

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	rf.apply_channel = applyCh

	// start ticker goroutine to start elections
	go rf.ticker()
	go rf.check_commit_and_try_apply()

	return rf
}
